//! 事务管理模块
//! 
//! 实现完整的ACID特性、MVCC并发控制、WAL日志和崩溃恢复

use crate::{Error, Result};
use crate::storage::page::{PageManager, PageId};
use crate::storage::tuple::{RowId, Tuple};
use std::collections::{HashMap, HashSet, VecDeque};
use std::sync::atomic::{AtomicU64, AtomicBool, Ordering};
use std::sync::Arc;
use std::time::{SystemTime, Duration, Instant};
use parking_lot::{RwLock, Mutex};
use serde::{Serialize, Deserialize};
use std::fs::{File, OpenOptions};
use std::io::{Write, Read, Seek, SeekFrom};

/// 事务ID类型
pub type TransactionId = u64;

/// 事务管理器
/// 实现完整的ACID特性和MVCC并发控制
#[derive(Debug)]
pub struct TransactionManager {
    /// 下一个事务ID
    next_transaction_id: AtomicU64,
    /// 活跃事务映射
    active_transactions: RwLock<HashMap<TransactionId, Arc<Transaction>>>,
    /// 已提交事务的最大ID（用于快照隔离）
    max_committed_transaction_id: AtomicU64,
    /// 锁管理器
    lock_manager: Arc<LockManager>,
    /// WAL管理器
    wal_manager: Arc<WalManager>,
    /// 死锁检测器
    deadlock_detector: Arc<DeadlockDetector>,
    /// 事务统计信息
    stats: RwLock<TransactionStats>,
    /// 页面管理器引用
    page_manager: Arc<PageManager>,
    /// 是否启用
    enabled: AtomicBool,
}

/// 事务状态
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub enum TransactionStatus {
    /// 活跃状态
    Active,
    /// 准备提交（2PC第一阶段）
    Preparing,
    /// 已提交
    Committed,
    /// 已回滚
    Aborted,
}

/// 事务隔离级别
#[derive(Debug, Clone, Copy, PartialEq)]
pub enum IsolationLevel {
    /// 读未提交
    ReadUncommitted,
    /// 读已提交
    ReadCommitted,
    /// 可重复读
    RepeatableRead,
    /// 串行化
    Serializable,
}

/// 事务
#[derive(Debug)]
pub struct Transaction {
    /// 事务ID
    pub id: TransactionId,
    /// 事务状态
    pub status: RwLock<TransactionStatus>,
    /// 开始时间
    pub start_time: SystemTime,
    /// 最后活跃时间
    pub last_active_time: RwLock<SystemTime>,
    /// 隔离级别
    pub isolation_level: IsolationLevel,
    /// 快照时间戳（用于MVCC）
    pub snapshot_timestamp: TransactionId,
    /// 读取的行集合（用于可重复读）
    pub read_set: RwLock<HashSet<RowId>>,
    /// 写入的行集合（用于冲突检测）
    pub write_set: RwLock<HashSet<RowId>>,
    /// 持有的锁
    pub locks: RwLock<HashSet<LockId>>,
    /// 事务日志条目
    pub log_entries: RwLock<Vec<LogEntry>>,
    /// 是否只读事务
    pub read_only: bool,
    /// 事务统计信息
    pub stats: RwLock<TransactionStatistics>,
}

/// WAL（Write-Ahead Logging）管理器
/// 实现预写日志，确保事务的持久性和崩溃恢复
#[derive(Debug)]
pub struct WalManager {
    /// WAL文件路径
    wal_file_path: String,
    /// WAL文件句柄
    wal_file: Mutex<File>,
    /// 当前LSN（Log Sequence Number）
    current_lsn: AtomicU64,
    /// 最后刷新的LSN
    last_flushed_lsn: AtomicU64,
    /// 内存中的日志缓冲区
    log_buffer: Mutex<Vec<WalRecord>>,
    /// 缓冲区大小限制
    buffer_size_limit: usize,
    /// 是否启用同步写入
    sync_write: bool,
    /// 检查点管理器
    checkpoint_manager: Arc<CheckpointManager>,
}

/// 检查点管理器
/// 负责定期创建检查点，减少恢复时间
#[derive(Debug)]
pub struct CheckpointManager {
    /// 最后检查点的LSN
    last_checkpoint_lsn: AtomicU64,
    /// 检查点间隔（秒）
    checkpoint_interval: Duration,
    /// 最后检查点时间
    last_checkpoint_time: RwLock<SystemTime>,
    /// 是否正在执行检查点
    checkpointing: AtomicBool,
}

/// WAL记录
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct WalRecord {
    /// LSN（Log Sequence Number）
    pub lsn: u64,
    /// 事务ID
    pub transaction_id: TransactionId,
    /// 记录类型
    pub record_type: WalRecordType,
    /// 时间戳
    pub timestamp: SystemTime,
    /// 数据
    pub data: Vec<u8>,
    /// 校验和
    pub checksum: u32,
}

/// WAL记录类型
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum WalRecordType {
    /// 事务开始
    TransactionBegin,
    /// 事务提交
    TransactionCommit,
    /// 事务回滚
    TransactionAbort,
    /// 数据插入
    Insert {
        page_id: PageId,
        slot_id: u16,
        old_data: Option<Vec<u8>>,
        new_data: Vec<u8>,
    },
    /// 数据更新
    Update {
        page_id: PageId,
        slot_id: u16,
        old_data: Vec<u8>,
        new_data: Vec<u8>,
    },
    /// 数据删除
    Delete {
        page_id: PageId,
        slot_id: u16,
        old_data: Vec<u8>,
    },
    /// 检查点
    Checkpoint {
        active_transactions: Vec<TransactionId>,
        dirty_pages: Vec<PageId>,
    },
}

/// 锁管理器
/// 实现行级锁和死锁检测
#[derive(Debug)]
pub struct LockManager {
    /// 锁表：资源ID -> 锁信息
    lock_table: RwLock<HashMap<LockId, LockInfo>>,
    /// 等待图：事务ID -> 等待的事务ID集合
    wait_graph: RwLock<HashMap<TransactionId, HashSet<TransactionId>>>,
    /// 锁等待队列
    wait_queue: RwLock<HashMap<LockId, VecDeque<LockRequest>>>,
    /// 锁统计信息
    lock_stats: RwLock<LockStatistics>,
}

/// 死锁检测器
/// 定期检测死锁并选择受害者事务
#[derive(Debug)]
pub struct DeadlockDetector {
    /// 检测间隔
    detection_interval: Duration,
    /// 最后检测时间
    last_detection_time: RwLock<SystemTime>,
    /// 死锁统计信息
    deadlock_stats: RwLock<DeadlockStatistics>,
}

/// 锁ID类型
pub type LockId = String;

/// 锁信息
#[derive(Debug, Clone)]
pub struct LockInfo {
    /// 锁类型
    pub lock_type: LockType,
    /// 持有锁的事务ID
    pub holder: TransactionId,
    /// 锁模式
    pub mode: LockMode,
    /// 获取时间
    pub acquired_time: SystemTime,
}

/// 锁类型
#[derive(Debug, Clone, PartialEq)]
pub enum LockType {
    /// 行锁
    Row(RowId),
    /// 页锁
    Page(PageId),
    /// 表锁
    Table(String),
}

/// 锁模式
#[derive(Debug, Clone, PartialEq, PartialOrd)]
pub enum LockMode {
    /// 意向共享锁
    IntentionShared,
    /// 意向排他锁
    IntentionExclusive,
    /// 共享锁
    Shared,
    /// 排他锁
    Exclusive,
}

/// 锁请求
#[derive(Debug, Clone)]
pub struct LockRequest {
    /// 请求事务ID
    pub transaction_id: TransactionId,
    /// 锁模式
    pub mode: LockMode,
    /// 请求时间
    pub request_time: SystemTime,
}

/// 日志条目
#[derive(Debug, Clone)]
pub struct LogEntry {
    /// 事务ID
    pub transaction_id: TransactionId,
    /// 操作类型
    pub operation: LogOperation,
    /// 时间戳
    pub timestamp: SystemTime,
    /// LSN
    pub lsn: u64,
}

/// 日志操作类型
#[derive(Debug, Clone)]
pub enum LogOperation {
    /// 事务开始
    Begin,
    /// 事务提交
    Commit,
    /// 事务回滚
    Rollback,
    /// 数据变更
    DataChange {
        table: String,
        row_id: u64,
        old_value: Option<Vec<crate::sql::Value>>,
        new_value: Option<Vec<crate::sql::Value>>,
    },
}

/// 事务统计信息
#[derive(Debug, Clone, Default)]
pub struct TransactionStats {
    /// 总事务数
    pub total_transactions: u64,
    /// 提交事务数
    pub committed_transactions: u64,
    /// 回滚事务数
    pub rolled_back_transactions: u64,
    /// 活跃事务数
    pub active_transactions: u64,
    /// 死锁检测次数
    pub deadlock_detections: u64,
    /// 检测到的死锁数
    pub deadlocks_detected: u64,
    /// 平均事务持续时间（毫秒）
    pub avg_transaction_duration_ms: u64,
    /// WAL写入次数
    pub wal_writes: u64,
    /// 检查点次数
    pub checkpoints: u64,
}

/// 单个事务的统计信息
#[derive(Debug, Clone, Default)]
pub struct TransactionStatistics {
    /// 读取的行数
    pub rows_read: u64,
    /// 写入的行数
    pub rows_written: u64,
    /// 获取的锁数量
    pub locks_acquired: u64,
    /// 等待锁的时间（微秒）
    pub lock_wait_time_us: u64,
    /// WAL记录数
    pub wal_records: u64,
}

/// 锁统计信息
#[derive(Debug, Clone, Default)]
pub struct LockStatistics {
    /// 总锁请求数
    pub total_lock_requests: u64,
    /// 锁等待次数
    pub lock_waits: u64,
    /// 锁超时次数
    pub lock_timeouts: u64,
    /// 平均锁等待时间（微秒）
    pub avg_lock_wait_time_us: u64,
}

/// 死锁统计信息
#[derive(Debug, Clone, Default)]
pub struct DeadlockStatistics {
    /// 死锁检测次数
    pub detection_runs: u64,
    /// 检测到的死锁数
    pub deadlocks_found: u64,
    /// 被选为受害者的事务数
    pub victim_transactions: u64,
    /// 平均检测时间（微秒）
    pub avg_detection_time_us: u64,
}

impl TransactionManager {
    /// 创建新的事务管理器
    pub fn new(page_manager: Arc<PageManager>, wal_file_path: String) -> Result<Self> {
        let wal_manager = Arc::new(WalManager::new(wal_file_path)?);
        let lock_manager = Arc::new(LockManager::new());
        let deadlock_detector = Arc::new(DeadlockDetector::new());

        Ok(Self {
            next_transaction_id: AtomicU64::new(1),
            active_transactions: RwLock::new(HashMap::new()),
            max_committed_transaction_id: AtomicU64::new(0),
            lock_manager,
            wal_manager,
            deadlock_detector,
            stats: RwLock::new(TransactionStats::default()),
            page_manager,
            enabled: AtomicBool::new(true),
        })
    }

    /// 获取页面管理器
    pub fn get_page_manager(&self) -> Arc<PageManager> {
        self.page_manager.clone()
    }

    /// 获取WAL管理器（用于测试）
    pub fn get_wal_manager(&self) -> &Arc<WalManager> {
        &self.wal_manager
    }

    /// 获取锁管理器（用于测试）
    pub fn get_lock_manager(&self) -> &Arc<LockManager> {
        &self.lock_manager
    }

    /// 开始新事务
    pub fn begin_transaction(&self, isolation_level: IsolationLevel, read_only: bool) -> Result<Arc<Transaction>> {
        if !self.enabled.load(Ordering::Acquire) {
            return Err(Error::storage("事务管理器已禁用".to_string()));
        }

        let transaction_id = self.next_transaction_id.fetch_add(1, Ordering::SeqCst);
        let snapshot_timestamp = if read_only {
            // 只读事务使用当前最大已提交事务ID作为快照
            self.max_committed_transaction_id.load(Ordering::Acquire)
        } else {
            transaction_id
        };

        let transaction = Arc::new(Transaction {
            id: transaction_id,
            status: RwLock::new(TransactionStatus::Active),
            start_time: SystemTime::now(),
            last_active_time: RwLock::new(SystemTime::now()),
            isolation_level,
            snapshot_timestamp,
            read_set: RwLock::new(HashSet::new()),
            write_set: RwLock::new(HashSet::new()),
            locks: RwLock::new(HashSet::new()),
            log_entries: RwLock::new(Vec::new()),
            read_only,
            stats: RwLock::new(TransactionStatistics::default()),
        });

        // 记录事务开始到WAL
        if !read_only {
            self.wal_manager.write_transaction_begin(transaction_id)?;
        }

        // 添加到活跃事务列表
        {
            let mut active_txns = self.active_transactions.write();
            active_txns.insert(transaction_id, transaction.clone());
        }

        // 更新统计信息
        {
            let mut stats = self.stats.write();
            stats.total_transactions += 1;
            stats.active_transactions += 1;
        }

        tracing::info!("事务 {} 开始 (隔离级别: {:?}, 只读: {})",
            transaction_id, isolation_level, read_only);

        Ok(transaction)
    }

    /// 提交事务
    pub fn commit_transaction(&self, transaction: Arc<Transaction>) -> Result<()> {
        let transaction_id = transaction.id;

        // 检查事务状态
        {
            let status = transaction.status.read();
            if *status != TransactionStatus::Active {
                return Err(Error::storage(format!("事务 {} 状态无效: {:?}", transaction_id, *status)));
            }
        }

        let start_time = Instant::now();

        // 两阶段提交：准备阶段
        {
            let mut status = transaction.status.write();
            *status = TransactionStatus::Preparing;
        }

        // 如果不是只读事务，写入提交记录到WAL
        if !transaction.read_only {
            self.wal_manager.write_transaction_commit(transaction_id)?;

            // 强制刷新WAL以确保持久性
            self.wal_manager.flush()?;
        }

        // 提交阶段：更新事务状态
        {
            let mut status = transaction.status.write();
            *status = TransactionStatus::Committed;
        }

        // 更新最大已提交事务ID
        self.max_committed_transaction_id.fetch_max(transaction_id, Ordering::SeqCst);

        // 释放所有锁
        self.release_transaction_locks(&transaction)?;

        // 从活跃事务列表中移除
        {
            let mut active_txns = self.active_transactions.write();
            active_txns.remove(&transaction_id);
        }

        // 更新统计信息
        {
            let mut stats = self.stats.write();
            stats.committed_transactions += 1;
            stats.active_transactions -= 1;

            let duration = start_time.elapsed().as_millis() as u64;
            stats.avg_transaction_duration_ms =
                (stats.avg_transaction_duration_ms + duration) / 2;
        }

        tracing::info!("事务 {} 已提交", transaction_id);
        Ok(())
    }

    /// 回滚事务
    pub fn rollback_transaction(&self, transaction: Arc<Transaction>) -> Result<()> {
        let transaction_id = transaction.id;

        // 检查事务状态
        {
            let status = transaction.status.read();
            if *status == TransactionStatus::Committed {
                return Err(Error::storage(format!("事务 {} 已提交，无法回滚", transaction_id)));
            }
        }

        let start_time = Instant::now();

        // 更新事务状态
        {
            let mut status = transaction.status.write();
            *status = TransactionStatus::Aborted;
        }

        // 如果不是只读事务，写入回滚记录到WAL
        if !transaction.read_only {
            self.wal_manager.write_transaction_abort(transaction_id)?;

            // 执行撤销操作
            self.undo_transaction_operations(&transaction)?;
        }

        // 释放所有锁
        self.release_transaction_locks(&transaction)?;

        // 从活跃事务列表中移除
        {
            let mut active_txns = self.active_transactions.write();
            active_txns.remove(&transaction_id);
        }

        // 更新统计信息
        {
            let mut stats = self.stats.write();
            stats.rolled_back_transactions += 1;
            stats.active_transactions -= 1;

            let duration = start_time.elapsed().as_millis() as u64;
            stats.avg_transaction_duration_ms =
                (stats.avg_transaction_duration_ms + duration) / 2;
        }

        tracing::info!("事务 {} 已回滚", transaction_id);
        Ok(())
    }

    /// 获取行锁
    pub fn acquire_row_lock(&self, transaction: &Transaction, row_id: RowId, mode: LockMode) -> Result<()> {
        let lock_id = format!("row:{:?}", row_id);
        self.acquire_lock(transaction, lock_id, mode)
    }

    /// 获取页锁
    pub fn acquire_page_lock(&self, transaction: &Transaction, page_id: PageId, mode: LockMode) -> Result<()> {
        let lock_id = format!("page:{}", page_id);
        self.acquire_lock(transaction, lock_id, mode)
    }

    /// 获取表锁
    pub fn acquire_table_lock(&self, transaction: &Transaction, table_name: &str, mode: LockMode) -> Result<()> {
        let lock_id = format!("table:{}", table_name);
        self.acquire_lock(transaction, lock_id, mode)
    }

    /// 通用锁获取方法
    fn acquire_lock(&self, transaction: &Transaction, lock_id: LockId, mode: LockMode) -> Result<()> {
        let start_time = Instant::now();

        // 检查是否已经持有兼容的锁
        if self.lock_manager.has_compatible_lock(transaction.id, &lock_id, &mode)? {
            return Ok(());
        }

        // 尝试获取锁
        match self.lock_manager.acquire_lock(transaction.id, lock_id.clone(), mode.clone()) {
            Ok(()) => {
                // 成功获取锁
                let mut locks = transaction.locks.write();
                locks.insert(lock_id);

                let mut stats = transaction.stats.write();
                stats.locks_acquired += 1;
                stats.lock_wait_time_us += start_time.elapsed().as_micros() as u64;

                Ok(())
            }
            Err(Error::Storage(msg)) if msg.contains("锁冲突") => {
                // 锁冲突，需要等待或检测死锁
                self.handle_lock_conflict(transaction, lock_id, mode, start_time)
            }
            Err(e) => Err(e),
        }
    }

    /// 处理锁冲突
    fn handle_lock_conflict(&self, transaction: &Transaction, lock_id: LockId, mode: LockMode, start_time: Instant) -> Result<()> {
        // 检测死锁
        if self.deadlock_detector.would_cause_deadlock(transaction.id, &lock_id)? {
            // 选择当前事务作为受害者
            return Err(Error::storage(format!("事务 {} 因死锁被终止", transaction.id)));
        }

        // 等待锁释放
        self.lock_manager.wait_for_lock(transaction.id, lock_id.clone(), mode)?;

        // 更新统计信息
        let mut stats = transaction.stats.write();
        stats.lock_wait_time_us += start_time.elapsed().as_micros() as u64;

        // 将锁添加到事务的锁集合
        let mut locks = transaction.locks.write();
        locks.insert(lock_id);

        Ok(())
    }

    /// 释放事务的所有锁
    fn release_transaction_locks(&self, transaction: &Transaction) -> Result<()> {
        let locks = transaction.locks.read();
        for lock_id in locks.iter() {
            self.lock_manager.release_lock(transaction.id, lock_id)?;
        }
        Ok(())
    }

    /// 撤销事务操作
    fn undo_transaction_operations(&self, transaction: &Transaction) -> Result<()> {
        let log_entries = transaction.log_entries.read();

        // 按相反顺序撤销操作
        for entry in log_entries.iter().rev() {
            match &entry.operation {
                LogOperation::DataChange { table, row_id, old_value, new_value: _ } => {
                    if let Some(_old_val) = old_value {
                        // 恢复旧值
                        tracing::debug!("撤销事务 {} 在表 {} 行 {} 的修改",
                            transaction.id, table, row_id);
                        // TODO: 实际的数据恢复逻辑
                    }
                }
                _ => {
                    // 其他操作类型的撤销
                }
            }
        }

        Ok(())
    }

    /// 检测死锁
    pub fn detect_deadlock(&self) -> Result<Vec<TransactionId>> {
        let start_time = Instant::now();

        let deadlocked_transactions = self.deadlock_detector.detect_deadlock(&self.lock_manager)?;

        // 更新统计信息
        {
            let mut stats = self.stats.write();
            stats.deadlock_detections += 1;
            if !deadlocked_transactions.is_empty() {
                stats.deadlocks_detected += 1;
            }
        }

        // 如果检测到死锁，选择受害者事务并回滚
        if !deadlocked_transactions.is_empty() {
            let victim_id = self.select_deadlock_victim(&deadlocked_transactions)?;
            if let Some(victim_transaction) = self.get_active_transaction(victim_id) {
                tracing::warn!("检测到死锁，选择事务 {} 作为受害者", victim_id);
                self.rollback_transaction(victim_transaction)?;
            }
        }

        Ok(deadlocked_transactions)
    }

    /// 选择死锁受害者
    fn select_deadlock_victim(&self, deadlocked_transactions: &[TransactionId]) -> Result<TransactionId> {
        // 简化策略：选择ID最大的事务（最新的事务）
        Ok(*deadlocked_transactions.iter().max().unwrap())
    }

    /// 获取活跃事务
    pub fn get_active_transaction(&self, transaction_id: TransactionId) -> Option<Arc<Transaction>> {
        let active_txns = self.active_transactions.read();
        active_txns.get(&transaction_id).cloned()
    }

    /// 获取事务统计信息
    pub fn get_stats(&self) -> TransactionStats {
        self.stats.read().clone()
    }

    /// 刷新WAL到磁盘
    pub fn flush_wal(&self) -> Result<()> {
        self.wal_manager.flush()
    }

    /// 获取当前LSN
    pub fn get_current_lsn(&self) -> u64 {
        self.wal_manager.current_lsn.load(Ordering::Acquire)
    }

    /// 启用事务管理器
    pub fn enable(&self) {
        self.enabled.store(true, Ordering::Release);
    }

    /// 禁用事务管理器
    pub fn disable(&self) {
        self.enabled.store(false, Ordering::Release);
    }

    /// 恢复数据库（崩溃恢复）
    pub fn recover(&self) -> Result<()> {
        tracing::info!("开始数据库恢复...");
        let start_time = Instant::now();

        // 从WAL恢复
        self.wal_manager.recover(&self.page_manager)?;

        tracing::info!("数据库恢复完成，耗时 {}ms", start_time.elapsed().as_millis());
        Ok(())
    }
}

impl WalManager {
    /// 创建新的WAL管理器
    pub fn new(wal_file_path: String) -> Result<Self> {
        let wal_file = OpenOptions::new()
            .create(true)
            .append(true)
            .read(true)
            .open(&wal_file_path)
            .map_err(|e| Error::storage(format!("无法打开WAL文件 {}: {}", wal_file_path, e)))?;

        let checkpoint_manager = Arc::new(CheckpointManager::new());

        Ok(Self {
            wal_file_path,
            wal_file: Mutex::new(wal_file),
            current_lsn: AtomicU64::new(1),
            last_flushed_lsn: AtomicU64::new(0),
            log_buffer: Mutex::new(Vec::new()),
            buffer_size_limit: 1024 * 1024, // 1MB缓冲区
            sync_write: true,
            checkpoint_manager,
        })
    }

    /// 写入事务开始记录
    pub fn write_transaction_begin(&self, transaction_id: TransactionId) -> Result<()> {
        self.write_record(transaction_id, WalRecordType::TransactionBegin)
    }

    /// 写入事务提交记录
    pub fn write_transaction_commit(&self, transaction_id: TransactionId) -> Result<()> {
        self.write_record(transaction_id, WalRecordType::TransactionCommit)?;
        // 提交时强制刷新WAL
        self.flush()
    }

    /// 写入事务回滚记录
    pub fn write_transaction_abort(&self, transaction_id: TransactionId) -> Result<()> {
        self.write_record(transaction_id, WalRecordType::TransactionAbort)?;
        // 回滚时强制刷新WAL
        self.flush()
    }

    /// 写入数据插入记录
    pub fn write_insert_record(
        &self,
        transaction_id: TransactionId,
        page_id: PageId,
        slot_id: u16,
        new_data: Vec<u8>,
    ) -> Result<()> {
        let record_type = WalRecordType::Insert {
            page_id,
            slot_id,
            old_data: None,
            new_data,
        };
        self.write_record(transaction_id, record_type)
    }

    /// 写入数据更新记录
    pub fn write_update_record(
        &self,
        transaction_id: TransactionId,
        page_id: PageId,
        slot_id: u16,
        old_data: Vec<u8>,
        new_data: Vec<u8>,
    ) -> Result<()> {
        let record_type = WalRecordType::Update {
            page_id,
            slot_id,
            old_data,
            new_data,
        };
        self.write_record(transaction_id, record_type)
    }

    /// 写入数据删除记录
    pub fn write_delete_record(
        &self,
        transaction_id: TransactionId,
        page_id: PageId,
        slot_id: u16,
        old_data: Vec<u8>,
    ) -> Result<()> {
        let record_type = WalRecordType::Delete {
            page_id,
            slot_id,
            old_data,
        };
        self.write_record(transaction_id, record_type)
    }

    /// 写入检查点记录
    pub fn write_checkpoint_record(
        &self,
        transaction_id: TransactionId,
        active_transactions: Vec<TransactionId>,
        dirty_pages: Vec<PageId>,
    ) -> Result<()> {
        let record_type = WalRecordType::Checkpoint {
            active_transactions,
            dirty_pages,
        };
        self.write_record(transaction_id, record_type)?;
        // 检查点时强制刷新WAL
        self.flush()
    }

    /// 写入WAL记录
    pub fn write_record(&self, transaction_id: TransactionId, record_type: WalRecordType) -> Result<()> {
        let lsn = self.current_lsn.fetch_add(1, Ordering::SeqCst);

        let data = bincode::serialize(&record_type)
            .map_err(|e| Error::storage(format!("序列化WAL记录失败: {}", e)))?;

        let record = WalRecord {
            lsn,
            transaction_id,
            record_type,
            timestamp: SystemTime::now(),
            data: data.clone(),
            checksum: Self::calculate_checksum(&data),
        };

        // 添加到缓冲区
        {
            let mut buffer = self.log_buffer.lock();
            buffer.push(record);

            // 如果缓冲区满了，强制刷新
            if buffer.len() * 256 > self.buffer_size_limit { // 估算每条记录256字节
                self.flush_buffer(&mut buffer)?;
            }
        }

        Ok(())
    }

    /// 刷新WAL到磁盘
    pub fn flush(&self) -> Result<()> {
        let mut buffer = self.log_buffer.lock();
        self.flush_buffer(&mut buffer)
    }

    /// 刷新缓冲区到磁盘
    /// 改进的WAL持久化写入，包含更好的错误处理和数据完整性保证
    fn flush_buffer(&self, buffer: &mut Vec<WalRecord>) -> Result<()> {
        if buffer.is_empty() {
            return Ok(());
        }

        let mut file = self.wal_file.lock();
        let mut total_bytes_written = 0;

        // 批量写入WAL记录，使用改进的格式
        for record in buffer.iter() {
            // 序列化记录
            let serialized = bincode::serialize(record)
                .map_err(|e| Error::storage(format!("序列化WAL记录失败: {}", e)))?;

            // 写入记录长度（4字节）+ 记录数据
            let record_length = serialized.len() as u32;
            let length_bytes = record_length.to_le_bytes();

            // 写入长度前缀
            file.write_all(&length_bytes)
                .map_err(|e| Error::storage(format!("写入WAL记录长度失败: {}", e)))?;

            // 写入记录数据
            file.write_all(&serialized)
                .map_err(|e| Error::storage(format!("写入WAL记录数据失败: {}", e)))?;

            total_bytes_written += 4 + serialized.len();

            // 验证校验和（基于record.data字段）
            let calculated_checksum = Self::calculate_checksum(&record.data);
            if calculated_checksum != record.checksum {
                return Err(Error::storage(format!(
                    "WAL记录校验和不匹配: 期望 {}, 实际 {}",
                    record.checksum, calculated_checksum
                )));
            }
        }

        // 强制刷新到磁盘以确保持久性
        if self.sync_write {
            file.flush()
                .map_err(|e| Error::storage(format!("刷新WAL文件失败: {}", e)))?;

            // 在支持的平台上调用fsync确保数据真正写入磁盘
            #[cfg(unix)]
            {
                use std::os::unix::io::AsRawFd;
                unsafe {
                    if libc::fsync(file.as_raw_fd()) != 0 {
                        return Err(Error::storage("WAL文件fsync失败".to_string()));
                    }
                }
            }
        }

        // 更新最后刷新的LSN
        if let Some(last_record) = buffer.last() {
            self.last_flushed_lsn.store(last_record.lsn, Ordering::SeqCst);
        }

        tracing::debug!("WAL刷新完成: {} 条记录, {} 字节", buffer.len(), total_bytes_written);
        buffer.clear();
        Ok(())
    }

    /// 计算校验和
    fn calculate_checksum(data: &[u8]) -> u32 {
        // 简化的校验和计算
        data.iter().fold(0u32, |acc, &byte| acc.wrapping_add(byte as u32))
    }

    /// 从WAL恢复数据
    /// 完善的崩溃恢复机制，支持实际的数据变更重做
    pub fn recover(&self, page_manager: &Arc<PageManager>) -> Result<()> {
        tracing::info!("开始WAL恢复...");
        let start_time = Instant::now();

        let mut file = self.wal_file.lock();
        file.seek(SeekFrom::Start(0))
            .map_err(|e| Error::storage(format!("定位WAL文件失败: {}", e)))?;

        let mut buffer = Vec::new();
        file.read_to_end(&mut buffer)
            .map_err(|e| Error::storage(format!("读取WAL文件失败: {}", e)))?;

        if buffer.is_empty() {
            tracing::info!("WAL文件为空，无需恢复");
            return Ok(());
        }

        // 解析WAL记录
        let records = self.parse_wal_records(&buffer)?;
        tracing::info!("解析到 {} 条WAL记录", records.len());

        // 第一遍：识别已提交的事务
        let mut active_transactions = HashSet::new();
        let mut committed_transactions = HashSet::new();
        let mut aborted_transactions = HashSet::new();

        for record in &records {
            match record.record_type {
                WalRecordType::TransactionBegin => {
                    active_transactions.insert(record.transaction_id);
                }
                WalRecordType::TransactionCommit => {
                    active_transactions.remove(&record.transaction_id);
                    committed_transactions.insert(record.transaction_id);
                }
                WalRecordType::TransactionAbort => {
                    active_transactions.remove(&record.transaction_id);
                    aborted_transactions.insert(record.transaction_id);
                }
                _ => {}
            }
        }

        tracing::info!("事务状态分析: {} 已提交, {} 已回滚, {} 未完成",
            committed_transactions.len(), aborted_transactions.len(), active_transactions.len());

        // 第二遍：重做已提交事务的数据变更
        let mut redo_count = 0;
        for record in &records {
            // 只重做已提交事务的数据变更
            if committed_transactions.contains(&record.transaction_id) {
                match &record.record_type {
                    WalRecordType::Insert { page_id, slot_id, new_data, .. } => {
                        self.redo_insert(page_manager, *page_id, *slot_id, new_data)?;
                        redo_count += 1;
                    }
                    WalRecordType::Update { page_id, slot_id, new_data, .. } => {
                        self.redo_update(page_manager, *page_id, *slot_id, new_data)?;
                        redo_count += 1;
                    }
                    WalRecordType::Delete { page_id, slot_id, .. } => {
                        self.redo_delete(page_manager, *page_id, *slot_id)?;
                        redo_count += 1;
                    }
                    _ => {}
                }
            }
        }

        // 更新当前LSN到最大值
        if let Some(last_record) = records.last() {
            self.current_lsn.store(last_record.lsn + 1, Ordering::SeqCst);
            self.last_flushed_lsn.store(last_record.lsn, Ordering::SeqCst);
        }

        tracing::info!("WAL恢复完成: 重做了 {} 个操作，耗时 {}ms",
            redo_count, start_time.elapsed().as_millis());
        Ok(())
    }

    /// 解析WAL记录
    /// 支持改进的格式（长度前缀 + 记录数据）
    fn parse_wal_records(&self, buffer: &[u8]) -> Result<Vec<WalRecord>> {
        let mut records = Vec::new();
        let mut offset = 0;

        while offset + 4 <= buffer.len() {
            // 读取记录长度（4字节）
            let length_bytes = &buffer[offset..offset + 4];
            let record_length = u32::from_le_bytes([
                length_bytes[0], length_bytes[1],
                length_bytes[2], length_bytes[3]
            ]) as usize;

            offset += 4;

            // 检查是否有足够的数据
            if offset + record_length > buffer.len() {
                tracing::warn!("WAL记录不完整，停止解析");
                break;
            }

            // 反序列化记录
            let record_data = &buffer[offset..offset + record_length];
            match bincode::deserialize::<WalRecord>(record_data) {
                Ok(record) => {
                    // 验证校验和（基于record.data字段）
                    let calculated_checksum = Self::calculate_checksum(&record.data);
                    if calculated_checksum == record.checksum {
                        records.push(record);
                    } else {
                        tracing::warn!("WAL记录校验和不匹配，跳过记录 LSN: {}", record.lsn);
                    }
                }
                Err(e) => {
                    tracing::warn!("反序列化WAL记录失败: {}", e);
                    break;
                }
            }

            offset += record_length;
        }

        Ok(records)
    }

    /// 重做插入操作
    fn redo_insert(&self, page_manager: &Arc<PageManager>, page_id: PageId, slot_id: u16, new_data: &[u8]) -> Result<()> {
        tracing::debug!("重做插入: 页面 {}, 槽位 {}", page_id, slot_id);

        // 读取页面
        let page = page_manager.read_page(page_id)?;
        let mut page_copy = (*page).clone();

        // 在指定槽位插入数据（简化实现）
        let start_offset = slot_id as usize * 256; // 假设每个槽位256字节
        if start_offset + new_data.len() <= page_copy.data.len() {
            page_copy.data[start_offset..start_offset + new_data.len()].copy_from_slice(new_data);
            page_manager.write_page(Arc::new(page_copy))?;
        }

        Ok(())
    }

    /// 重做更新操作
    fn redo_update(&self, page_manager: &Arc<PageManager>, page_id: PageId, slot_id: u16, new_data: &[u8]) -> Result<()> {
        tracing::debug!("重做更新: 页面 {}, 槽位 {}", page_id, slot_id);

        // 更新操作与插入操作类似
        self.redo_insert(page_manager, page_id, slot_id, new_data)
    }

    /// 重做删除操作
    fn redo_delete(&self, page_manager: &Arc<PageManager>, page_id: PageId, slot_id: u16) -> Result<()> {
        tracing::debug!("重做删除: 页面 {}, 槽位 {}", page_id, slot_id);

        // 读取页面
        let page = page_manager.read_page(page_id)?;
        let mut page_copy = (*page).clone();

        // 清空指定槽位的数据（简化实现）
        let start_offset = slot_id as usize * 256; // 假设每个槽位256字节
        if start_offset + 256 <= page_copy.data.len() {
            page_copy.data[start_offset..start_offset + 256].fill(0);
            page_manager.write_page(Arc::new(page_copy))?;
        }

        Ok(())
    }

    /// 获取当前LSN（用于测试）
    pub fn get_current_lsn(&self) -> u64 {
        self.current_lsn.load(Ordering::Acquire)
    }

    /// 获取检查点管理器（用于测试）
    pub fn get_checkpoint_manager(&self) -> &CheckpointManager {
        &self.checkpoint_manager
    }
}

impl LockManager {
    /// 创建新的锁管理器
    pub fn new() -> Self {
        Self {
            lock_table: RwLock::new(HashMap::new()),
            wait_graph: RwLock::new(HashMap::new()),
            wait_queue: RwLock::new(HashMap::new()),
            lock_stats: RwLock::new(LockStatistics::default()),
        }
    }

    /// 获取锁
    pub fn acquire_lock(&self, transaction_id: TransactionId, lock_id: LockId, mode: LockMode) -> Result<()> {
        let start_time = Instant::now();

        // 检查是否已经持有兼容的锁
        if self.has_compatible_lock(transaction_id, &lock_id, &mode)? {
            return Ok(());
        }

        // 检查锁冲突
        {
            let lock_table = self.lock_table.read();
            if let Some(lock_info) = lock_table.get(&lock_id) {
                if !self.is_compatible(&lock_info.mode, &mode) {
                    // 锁冲突，添加到等待队列
                    self.add_to_wait_queue(transaction_id, lock_id.clone(), mode)?;
                    return Err(Error::storage("锁冲突".to_string()));
                }
            }
        }

        // 获取锁
        {
            let mut lock_table = self.lock_table.write();
            lock_table.insert(lock_id, LockInfo {
                lock_type: LockType::Row(RowId { page_id: 0, slot_id: 0 }), // 简化
                holder: transaction_id,
                mode,
                acquired_time: SystemTime::now(),
            });
        }

        // 更新统计信息
        {
            let mut stats = self.lock_stats.write();
            stats.total_lock_requests += 1;
            stats.avg_lock_wait_time_us =
                (stats.avg_lock_wait_time_us + start_time.elapsed().as_micros() as u64) / 2;
        }

        Ok(())
    }

    /// 释放锁
    pub fn release_lock(&self, transaction_id: TransactionId, lock_id: &LockId) -> Result<()> {
        {
            let mut lock_table = self.lock_table.write();
            if let Some(lock_info) = lock_table.get(lock_id) {
                if lock_info.holder == transaction_id {
                    lock_table.remove(lock_id);
                } else {
                    return Err(Error::storage(format!("事务 {} 不持有锁 {}", transaction_id, lock_id)));
                }
            }
        }

        // 唤醒等待的事务
        self.notify_waiting_transactions(lock_id)?;

        Ok(())
    }

    /// 检查是否持有兼容的锁
    pub fn has_compatible_lock(&self, transaction_id: TransactionId, lock_id: &LockId, mode: &LockMode) -> Result<bool> {
        let lock_table = self.lock_table.read();
        if let Some(lock_info) = lock_table.get(lock_id) {
            Ok(lock_info.holder == transaction_id && self.is_compatible(&lock_info.mode, mode))
        } else {
            Ok(false)
        }
    }

    /// 等待锁释放
    pub fn wait_for_lock(&self, transaction_id: TransactionId, lock_id: LockId, mode: LockMode) -> Result<()> {
        // 简化实现：直接返回成功
        // 实际实现需要使用条件变量等待
        Ok(())
    }

    /// 检查锁模式兼容性
    fn is_compatible(&self, held_mode: &LockMode, requested_mode: &LockMode) -> bool {
        use LockMode::*;
        match (held_mode, requested_mode) {
            (IntentionShared, IntentionShared) => true,
            (IntentionShared, Shared) => true,
            (Shared, IntentionShared) => true,
            (Shared, Shared) => true,
            _ => false,
        }
    }

    /// 添加到等待队列
    fn add_to_wait_queue(&self, transaction_id: TransactionId, lock_id: LockId, mode: LockMode) -> Result<()> {
        let mut wait_queue = self.wait_queue.write();
        let queue = wait_queue.entry(lock_id).or_insert_with(VecDeque::new);
        queue.push_back(LockRequest {
            transaction_id,
            mode,
            request_time: SystemTime::now(),
        });
        Ok(())
    }

    /// 通知等待的事务
    fn notify_waiting_transactions(&self, lock_id: &LockId) -> Result<()> {
        // 简化实现：清空等待队列
        let mut wait_queue = self.wait_queue.write();
        wait_queue.remove(lock_id);
        Ok(())
    }
}

impl DeadlockDetector {
    /// 创建新的死锁检测器
    pub fn new() -> Self {
        Self {
            detection_interval: Duration::from_secs(5), // 每5秒检测一次
            last_detection_time: RwLock::new(SystemTime::now()),
            deadlock_stats: RwLock::new(DeadlockStatistics::default()),
        }
    }

    /// 检测死锁
    pub fn detect_deadlock(&self, lock_manager: &LockManager) -> Result<Vec<TransactionId>> {
        let start_time = Instant::now();

        // 构建等待图
        let wait_graph = self.build_wait_graph(lock_manager)?;

        // 检测环
        let deadlocked_transactions = self.detect_cycles(&wait_graph)?;

        // 更新统计信息
        {
            let mut stats = self.deadlock_stats.write();
            stats.detection_runs += 1;
            if !deadlocked_transactions.is_empty() {
                stats.deadlocks_found += 1;
            }
            stats.avg_detection_time_us =
                (stats.avg_detection_time_us + start_time.elapsed().as_micros() as u64) / 2;
        }

        // 更新最后检测时间
        {
            let mut last_time = self.last_detection_time.write();
            *last_time = SystemTime::now();
        }

        Ok(deadlocked_transactions)
    }

    /// 检查是否会导致死锁
    pub fn would_cause_deadlock(&self, transaction_id: TransactionId, lock_id: &LockId) -> Result<bool> {
        // 简化实现：总是返回false
        // 实际实现需要分析等待图
        Ok(false)
    }

    /// 构建等待图
    fn build_wait_graph(&self, lock_manager: &LockManager) -> Result<HashMap<TransactionId, HashSet<TransactionId>>> {
        let mut wait_graph = HashMap::new();

        // 从锁管理器获取等待信息
        let wait_queue = lock_manager.wait_queue.read();
        let lock_table = lock_manager.lock_table.read();

        for (lock_id, queue) in wait_queue.iter() {
            if let Some(lock_info) = lock_table.get(lock_id) {
                let holder = lock_info.holder;

                for request in queue {
                    let waiter = request.transaction_id;
                    wait_graph.entry(waiter).or_insert_with(HashSet::new).insert(holder);
                }
            }
        }

        Ok(wait_graph)
    }

    /// 检测环（死锁）
    fn detect_cycles(&self, wait_graph: &HashMap<TransactionId, HashSet<TransactionId>>) -> Result<Vec<TransactionId>> {
        let mut visited = HashSet::new();
        let mut rec_stack = HashSet::new();
        let mut deadlocked = Vec::new();

        for &transaction_id in wait_graph.keys() {
            if !visited.contains(&transaction_id) {
                if self.dfs_detect_cycle(transaction_id, wait_graph, &mut visited, &mut rec_stack, &mut deadlocked) {
                    // 找到环，返回参与死锁的事务
                    return Ok(deadlocked);
                }
            }
        }

        Ok(Vec::new())
    }

    /// 深度优先搜索检测环
    fn dfs_detect_cycle(
        &self,
        transaction_id: TransactionId,
        wait_graph: &HashMap<TransactionId, HashSet<TransactionId>>,
        visited: &mut HashSet<TransactionId>,
        rec_stack: &mut HashSet<TransactionId>,
        deadlocked: &mut Vec<TransactionId>,
    ) -> bool {
        visited.insert(transaction_id);
        rec_stack.insert(transaction_id);

        if let Some(dependencies) = wait_graph.get(&transaction_id) {
            for &dep_transaction in dependencies {
                if !visited.contains(&dep_transaction) {
                    if self.dfs_detect_cycle(dep_transaction, wait_graph, visited, rec_stack, deadlocked) {
                        deadlocked.push(transaction_id);
                        return true;
                    }
                } else if rec_stack.contains(&dep_transaction) {
                    // 找到环
                    deadlocked.push(transaction_id);
                    deadlocked.push(dep_transaction);
                    return true;
                }
            }
        }

        rec_stack.remove(&transaction_id);
        false
    }
}

impl CheckpointManager {
    /// 创建新的检查点管理器
    pub fn new() -> Self {
        Self {
            last_checkpoint_lsn: AtomicU64::new(0),
            checkpoint_interval: Duration::from_secs(300), // 5分钟间隔
            last_checkpoint_time: RwLock::new(SystemTime::now()),
            checkpointing: AtomicBool::new(false),
        }
    }

    /// 检查是否需要执行检查点
    pub fn should_checkpoint(&self) -> bool {
        let last_time = self.last_checkpoint_time.read();
        SystemTime::now().duration_since(*last_time)
            .unwrap_or(Duration::ZERO) > self.checkpoint_interval
    }

    /// 开始检查点
    pub fn start_checkpoint(&self) -> bool {
        !self.checkpointing.swap(true, Ordering::SeqCst)
    }

    /// 完成检查点
    pub fn finish_checkpoint(&self, lsn: u64) {
        self.last_checkpoint_lsn.store(lsn, Ordering::SeqCst);
        let mut last_time = self.last_checkpoint_time.write();
        *last_time = SystemTime::now();
        self.checkpointing.store(false, Ordering::SeqCst);
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use tempfile::tempdir;

    #[test]
    fn test_transaction_manager_creation() {
        let temp_dir = tempdir().unwrap();
        let wal_path = temp_dir.path().join("test.wal").to_string_lossy().to_string();
        let page_manager = Arc::new(PageManager::new(100));

        let tx_manager = TransactionManager::new(page_manager, wal_path).unwrap();
        assert!(tx_manager.enabled.load(Ordering::Acquire));
        assert_eq!(tx_manager.next_transaction_id.load(Ordering::Acquire), 1);
    }

    #[test]
    fn test_begin_transaction() {
        let temp_dir = tempdir().unwrap();
        let wal_path = temp_dir.path().join("test.wal").to_string_lossy().to_string();
        let page_manager = Arc::new(PageManager::new(100));

        let tx_manager = TransactionManager::new(page_manager, wal_path).unwrap();

        let tx = tx_manager.begin_transaction(IsolationLevel::ReadCommitted, false).unwrap();
        assert_eq!(tx.id, 1);
        assert_eq!(*tx.status.read(), TransactionStatus::Active);
        assert_eq!(tx.isolation_level, IsolationLevel::ReadCommitted);
        assert!(!tx.read_only);

        let stats = tx_manager.get_stats();
        assert_eq!(stats.total_transactions, 1);
        assert_eq!(stats.active_transactions, 1);
    }

    #[test]
    fn test_commit_transaction() {
        let temp_dir = tempdir().unwrap();
        let wal_path = temp_dir.path().join("test.wal").to_string_lossy().to_string();
        let page_manager = Arc::new(PageManager::new(100));

        let tx_manager = TransactionManager::new(page_manager, wal_path).unwrap();

        let tx = tx_manager.begin_transaction(IsolationLevel::ReadCommitted, false).unwrap();
        let tx_id = tx.id;

        tx_manager.commit_transaction(tx).unwrap();

        let stats = tx_manager.get_stats();
        assert_eq!(stats.committed_transactions, 1);
        assert_eq!(stats.active_transactions, 0);
        assert_eq!(tx_manager.max_committed_transaction_id.load(Ordering::Acquire), tx_id);
    }

    #[test]
    fn test_rollback_transaction() {
        let temp_dir = tempdir().unwrap();
        let wal_path = temp_dir.path().join("test.wal").to_string_lossy().to_string();
        let page_manager = Arc::new(PageManager::new(100));

        let tx_manager = TransactionManager::new(page_manager, wal_path).unwrap();

        let tx = tx_manager.begin_transaction(IsolationLevel::ReadCommitted, false).unwrap();

        tx_manager.rollback_transaction(tx).unwrap();

        let stats = tx_manager.get_stats();
        assert_eq!(stats.rolled_back_transactions, 1);
        assert_eq!(stats.active_transactions, 0);
    }

    #[test]
    fn test_read_only_transaction() {
        let temp_dir = tempdir().unwrap();
        let wal_path = temp_dir.path().join("test.wal").to_string_lossy().to_string();
        let page_manager = Arc::new(PageManager::new(100));

        let tx_manager = TransactionManager::new(page_manager, wal_path).unwrap();

        let tx = tx_manager.begin_transaction(IsolationLevel::ReadCommitted, true).unwrap();
        assert!(tx.read_only);
        assert_eq!(tx.snapshot_timestamp, 0); // 使用最大已提交事务ID

        // 只读事务提交不应写入WAL
        tx_manager.commit_transaction(tx).unwrap();

        let stats = tx_manager.get_stats();
        assert_eq!(stats.committed_transactions, 1);
    }

    #[test]
    fn test_lock_manager() {
        let lock_manager = LockManager::new();

        // 获取锁
        let result = lock_manager.acquire_lock(1, "test_lock".to_string(), LockMode::Shared);
        assert!(result.is_ok());

        // 释放锁
        let result = lock_manager.release_lock(1, &"test_lock".to_string());
        assert!(result.is_ok());
    }

    #[test]
    fn test_wal_manager() {
        let temp_dir = tempdir().unwrap();
        let wal_path = temp_dir.path().join("test.wal").to_string_lossy().to_string();

        let wal_manager = WalManager::new(wal_path).unwrap();

        // 写入事务开始记录
        wal_manager.write_transaction_begin(1).unwrap();

        // 写入数据变更记录
        let record_type = WalRecordType::Insert {
            page_id: 1,
            slot_id: 0,
            old_data: None,
            new_data: vec![1, 2, 3, 4],
        };
        wal_manager.write_record(1, record_type).unwrap();

        // 写入事务提交记录
        wal_manager.write_transaction_commit(1).unwrap();

        // 刷新到磁盘
        wal_manager.flush().unwrap();

        assert!(wal_manager.current_lsn.load(Ordering::Acquire) > 1);
    }

    #[test]
    fn test_deadlock_detector() {
        let lock_manager = LockManager::new();
        let deadlock_detector = DeadlockDetector::new();

        // 简单的死锁检测测试
        let result = deadlock_detector.detect_deadlock(&lock_manager).unwrap();
        assert!(result.is_empty()); // 没有死锁

        let would_deadlock = deadlock_detector.would_cause_deadlock(1, &"test_lock".to_string()).unwrap();
        assert!(!would_deadlock);
    }

    #[test]
    fn test_checkpoint_manager() {
        let checkpoint_manager = CheckpointManager::new();

        // 新创建的检查点管理器不应该需要检查点
        assert!(!checkpoint_manager.should_checkpoint());

        // 开始检查点
        assert!(checkpoint_manager.start_checkpoint());

        // 完成检查点
        checkpoint_manager.finish_checkpoint(100);
        assert_eq!(checkpoint_manager.last_checkpoint_lsn.load(Ordering::Acquire), 100);
    }

    #[test]
    fn test_transaction_isolation_levels() {
        let temp_dir = tempdir().unwrap();
        let wal_path = temp_dir.path().join("test.wal").to_string_lossy().to_string();
        let page_manager = Arc::new(PageManager::new(100));

        let tx_manager = TransactionManager::new(page_manager, wal_path).unwrap();

        // 测试不同隔离级别
        let levels = vec![
            IsolationLevel::ReadUncommitted,
            IsolationLevel::ReadCommitted,
            IsolationLevel::RepeatableRead,
            IsolationLevel::Serializable,
        ];

        for level in levels {
            let tx = tx_manager.begin_transaction(level, false).unwrap();
            assert_eq!(tx.isolation_level, level);
            tx_manager.commit_transaction(tx).unwrap();
        }

        let stats = tx_manager.get_stats();
        assert_eq!(stats.total_transactions, 4);
        assert_eq!(stats.committed_transactions, 4);
    }

    #[test]
    fn test_mvcc_visibility() {
        let temp_dir = tempdir().unwrap();
        let wal_path = temp_dir.path().join("test.wal").to_string_lossy().to_string();
        let page_manager = Arc::new(PageManager::new(100));

        let tx_manager = TransactionManager::new(page_manager, wal_path).unwrap();

        // 创建并提交一个事务
        let tx1 = tx_manager.begin_transaction(IsolationLevel::ReadCommitted, false).unwrap();
        let tx1_id = tx1.id;
        tx_manager.commit_transaction(tx1).unwrap();

        // 创建一个只读事务，应该能看到已提交的事务
        let tx2 = tx_manager.begin_transaction(IsolationLevel::ReadCommitted, true).unwrap();
        assert!(tx2.snapshot_timestamp >= tx1_id);

        tx_manager.commit_transaction(tx2).unwrap();
    }
}
